package com.lichao.sink;

import com.lichao.source.Event;
import com.lichao.utils.ReadProperties;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SinkToClickhouseTest {
    public static void main(String[] args) throws Exception {
        ReadProperties.init();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("test1", "./home", 1000L),
                new Event("test1", "./cart", 2000L),
                new Event("test1", "./prod?id=100", 3000L),
                new Event("test2", "./prod?id=200", 3500L),
                new Event("test2", "./prod?id=2", 2500L),
                new Event("test2", "./prod?id=300", 3600L),
                new Event("test2", "./home", 3000L),
                new Event("test3", "./prod?id=1", 2300L),
                new Event("test3", "./prod?id=3", 3300L));

        DataStreamSink<Event> eventDataStreamSink = stream.addSink(new SinkToClickhouse());
        env.execute();
    }
}
